// This file is part of OpenTSDB.
// Copyright (C) 2017 TeleDB Development Group.
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or (at your
// option) any later version.  This program is distributed in the hope that it
// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
// General Public License for more details.  You should have received a copy
// of the GNU Lesser General Public License along with this program.  If not,
// see <http://www.gnu.org/licenses/>.
package net.opentsdb.tsd.service;

import com.google.common.base.Strings;
import com.google.common.cache.Cache;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.stumbleupon.async.Callback;
import com.stumbleupon.async.Deferred;
import net.opentsdb.core.*;
import net.opentsdb.query.filter.TagVFilter;
import net.opentsdb.tools.http.SendClient;
import net.opentsdb.tsd.HttpQuery;
import net.opentsdb.tsd.http.ServiceResponse;
import net.opentsdb.uid.UniqueId;
import net.opentsdb.utils.ExpireFuture;
import net.opentsdb.utils.TimestampUtils;
import net.opentsdb.utils.hbase.HbaseTsdbUtils;
import net.opentsdb.utils.status.ClearDataStatus;
import org.hbase.async.DeleteRequest;
import org.hbase.async.KeyOnlyFilter;
import org.hbase.async.KeyValue;
import org.hbase.async.Scanner;
import org.jboss.netty.handler.codec.http.HttpResponseStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.LongAdder;
import java.util.stream.Collectors;

/**
 *  删除数据请求
 */
public class DeleteDataService extends RpcServiceImpl{

    private static final Logger LOG = LoggerFactory.getLogger(DeleteDataService.class);

    private HbaseTsdbUtils tsdbUtils;
    private final TSQuery dataQuery;

    private LongAdder count = new LongAdder();
    private Set<String> keys = Sets.newHashSet();

    private String jobId;

    public DeleteDataService(TSDB tsdb, HttpQuery query){
        super(tsdb, query);
        count.add(0L);
        tsdbUtils = tsdb.getHbaseWrapper().tsdbUtils;
        dataQuery = getQuery().serializer().parseQueryV1();
    }

    @Override
    public void doPost() {
        doDelete();
    }

    @Override
    public void doDelete() {
        Cache<String, String> cache = tsdb.getClearDataCache();

        jobId = getQuery().getQueryStringParam("jobId");

        if(Strings.isNullOrEmpty(jobId)){
            response = ServiceResponse
                    .builder(HttpResponseStatus.NOT_ACCEPTABLE)
                    .message("jobId con't be null")
                    .build();
            return;
        }

        cachePut(cache, jobId, ClearDataStatus.JobStatus.JOB_STATUS_RUNNING, 0L, keys);
        // 过滤器
        List<TagVFilter> filters = new ArrayList<>();
        dataQuery.getQueries().forEach(query->filters.addAll(query.getFilters()));

        try{
            dataQuery.validateAndSetQuery();
            IQuery[] queries = dataQuery.buildQueriesAsync(tsdb).join();

            CompletableFuture<Long> completableFuture =
                    CompletableFuture.completedFuture(queries)
                            .thenApplyAsync(query->{
                                try{
                                    return deleteData(query, dataQuery, filters);
                                }catch (Exception e){
                                    LOG.error("delete data error:{}",jobId,e);
                                    return ClearDataStatus.CLEAR_TIMEOUT*1000L;
                                }finally {}
                            });

            final CompletableFuture<Long> responseFuture = ExpireFuture.within(
                    completableFuture,
                    Duration.ofSeconds(ClearDataStatus.CLEAR_TIMEOUT));

            // TODO 试过很多方法，发现都并不能打断异步任务，后续再看了
            responseFuture.thenAccept(time->{
                cachePut(cache, jobId, ClearDataStatus.JobStatus.JOB_STATUS_FINISH, count.sum(), keys);
                /*sendResult(jobId,
                        ClearDataStatus.JobStatus.JOB_STATUS_FINISH.get(),
                        count.sum());*/
                // 删除成功，返回数据
                LOG.debug("send clear data message");
            }).exceptionally(throwable -> {
                // 超时处理
                boolean cancel = completableFuture.cancel(true);
                cachePut(cache, jobId, ClearDataStatus.JobStatus.JOB_STATUS_TIMEOUT, count.sum(), keys);
                // 超时返回删除信息
                sendResult(jobId,
                        ClearDataStatus.JobStatus.JOB_STATUS_TIMEOUT.get(),
                        count.sum(),
                        keys);

                LOG.error("timeout handle delete hbase data, cancel result:",cancel);
               return null;
            });

            response = ServiceResponse.builder(HttpResponseStatus.ACCEPTED).build();
        } catch (Exception e) {
            LOG.debug("delete data error",e);
            putErrorResponse(false, "delete metric data error");
        }
    }

    @Override
    public void checks() { }

    @Override
    public Object doParse(){
        // 解析放入#doDelete内
        return null;
    }

    /**
     * 删除数据
     *
     * @param queries
     * @param dataQuery
     * @param filters
     * @return
     * @throws Exception
     */
    private long deleteData(IQuery[] queries, TSQuery dataQuery, List<TagVFilter> filters) throws Exception{

        long startTime = System.currentTimeMillis();
        final List<Deferred<Object>> delGroup = new ArrayList<>();

        for (final IQuery query : queries) {
            final List<Scanner> scanners = Internal.getScanners(query);
            for (Scanner scanner : scanners) {
                ArrayList<ArrayList<KeyValue>> rows;
                scanner.setFilter(new KeyOnlyFilter());
                // 时间以小时为单位取整，提升扫描效率
                scanner.setTimeRange(
                        TimestampUtils.floorHours(dataQuery.startTime()),
                        TimestampUtils.floorHours(dataQuery.endTime()));

                while ((rows = scanner.nextRows().joinUninterruptibly()) != null) {
                    for (final ArrayList<KeyValue> row : rows) {
                        if(row != null && row.size()>0){

                            final byte[] key = row.get(0).key();

                            if(filters.size() > 0){
                                // filters过滤
                                Tags.getTagsAsync(tsdb, key).addCallback(tags -> {

                                    boolean flag = true;

                                    for (final TagVFilter filter : filters) {
                                        flag = filter.match(tags).joinUninterruptibly();
                                        if(!flag) break;
                                    }

                                    if(flag) delGroup.add(deleteAndCount(row));
                                    return null;
                                });
                            }else{
                                delGroup.add(deleteAndCount(row));
                            }
                        }
                    }
                }
            }
        }

        // 返回删除统计数据
        Deferred.group(delGroup)
                .addCallback(arg -> {
                    sendResult(jobId,
                            ClearDataStatus.JobStatus.JOB_STATUS_FINISH.get(),
                            count.sum(),
                            keys);
                    return null;
                });

        LOG.debug("delete data finish:{}",System.currentTimeMillis()-startTime);
        return System.currentTimeMillis()-startTime;
    }

    /**
     * 发送数据
     *
     * @param jobId     任务id
     * @param jobRs     任务状态
     * @param dps       清理总数
     */
    private void sendResult(String jobId, String jobRs, long dps, Set<String> tsuids){
        String url = tsdb.getBaseUrl().concat(tsdb.getConfig().clearDataPath());
        Map<String,Object> map = Maps.newHashMap();
        map.put("jobId",jobId);
        map.put("jobRs",jobRs);
        map.put("dps",String.valueOf(dps));

        if(dataQuery.getAll()){
            map.put("tsuids",tsuids);
        }else {
            try{
                map.put("tsuids",tsdbUtils.getDelTsuids(tsuids, tsdb, dataQuery.startTime(), dataQuery.endTime()));
            }catch (Exception e){
                LOG.error("build query or get delTsuids error:",e);
            }
        }

        try(SendClient client = new SendClient(url)){
            client.sendObject(map);
        }catch (Exception e){
            LOG.error("send url {} data failed",url,e);
        }
    }

    private void cachePut(Cache cache, String jobId, ClearDataStatus.JobStatus status, long count, Set<String> tsuids){
        // TODO 这里可以优化
        cache.put(ClearDataStatus.JOB_ID, jobId);
        cache.put(ClearDataStatus.JOB_STATUS, status.get());
        cache.put(ClearDataStatus.CLEAR_COUNT, String.valueOf(count));
        cache.put(ClearDataStatus.TSUIDS, String.valueOf(tsuids));
    }

    /**
     * 删除数据，统计删除数据
     *
     * @param rows
     */
    private Deferred<Object> deleteAndCount(ArrayList<KeyValue> rows){
        byte[] key = rows.get(0).key();
        final DeleteRequest del = new DeleteRequest(tsdb.dataTable(), key);

        // 返回删除的时间线
        final byte[] tsuid = UniqueId.getTSUIDFromKey(key, TSDB.metrics_width(), Const.TIMESTAMP_BYTES);
        keys.add(UniqueId.uidToString(tsuid));

        // TODO 写入的时间戳为秒计数会有问题
        for(KeyValue keyValue: rows){
            if(keyValue.qualifier().length < Const.TIMESTAMP_BYTES
                    && keyValue.qualifier().length > 0){
                count.increment();
            }else {
                count.add(keyValue.qualifier().length/Const.TIMESTAMP_BYTES);
            }
        }

        // TODO 数据量小会出现未删完就返回的情况
        return tsdb.getClient().delete(del);
    }

}